package com.example.redis.common.doJob;

import cn.hutool.core.date.DateUtil;
import com.example.redis.common.enums.RedisDelayQueueEnum;
import com.example.redis.handle.RedisDelayQueueHandle;
import com.example.redis.util.RedisDelayQueueUtil;
import com.example.redis.util.SpringUtils;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 启动延迟队列监测扫描
 *
 * @Author guoyb
 * @Date 2022/6/28  12:18
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {

    @Autowired
    private RedisDelayQueueUtil redisDelayQueueUtil;

    @Autowired
    private SpringUtils springUtils;

    @Autowired
    private ThreadPoolTaskExecutor ptask;


    @Override
    public void run(String... args) throws Exception {

        ThreadPoolExecutor executorService = new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(1000), new ThreadFactoryBuilder().setNameFormat("order-delay-%d").build());

        // RBlockingQueue blockingQueue = client.getBlockingQueue(jobsTag);
        // //下面2行 必不可少 防止出现 服务器重启后，延迟队列take数据阻塞，不执行，必须等到下一个内容offer时，队列才会把阻塞的消息全部处理掉
        // RDelayedQueue<DelayJob> delayedQueue = client.getDelayedQueue(blockingQueue);
        // delayedQueue.offer(null, 1, TimeUnit.SECONDS);

        ptask.execute(() -> {
            while (true) {
                try {
                    RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
                    for (RedisDelayQueueEnum queueEnum : queueEnums) {
                        Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
                        if (value != null) {
                            log.info("延迟队列监测扫描------{}", DateUtil.now());

                            RedisDelayQueueHandle<Object> redisDelayQueueHandle = springUtils.getBean(queueEnum.getBeanId());
                            executorService.execute(() -> {
                                redisDelayQueueHandle.execute(value);
                            });
                        }
                    }
                    //线程睡眠500毫秒
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    log.error("(Redission延迟队列监测异常中断) {}", e.getMessage());
                }
            }
        });
        log.info("(Redission延迟队列监测启动成功)");

    }

    /**
     * 启动线程获取队列*
     *
     * @param queueName                 queueName
     * @param redisDelayedQueueListener 任务回调监听
     * @param <T>                       泛型
     * @return
     */
    // private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
    //     RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
    //     //服务重启后，无offer，take不到信息。
    //     redissonClient.getDelayedQueue(blockingFairQueue);
    //     //由于此线程需要常驻，可以新建线程，不用交给线程池管理
    //     Thread thread = new Thread(() -> {
    //         logger.info("启动监听队列线程" + queueName);
    //         while (true) {
    //             try {
    //                 T t = blockingFairQueue.take();
    //                 logger.info("监听队列线程，监听名称：{},内容:{}", queueName, t);
    //                 redisDelayedQueueListener.invoke(t);
    //             } catch (Exception e) {
    //                 logger.info("监听队列线程错误,", e);
    //             }
    //         }
    //     });
    //     thread.setName(queueName);
    //     thread.start();
    // }
    //


}
